package com.yl.test;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBPoint;

import java.util.HashMap;
import java.util.List;

/**
 * @author wlf
 * @since 2022/8/1
 */
public class ObjMapFunc extends RichMapFunction<List<Object>, InfluxDBPoint> {

    private String tbl;

    public ObjMapFunc(String tbl) {
        this.tbl = tbl;
    }

    @Override
    public InfluxDBPoint map(List<Object> objectList) throws Exception {
        int i = 0;
        // 数据时间
        long timestamp = Long.parseLong(objectList.get(i++).toString());

        // tags
        HashMap<String, String> tags = new HashMap<>();
        tags.put("tid", objectList.get(i++).toString());

        // fields
        HashMap<String, Object> fields = new HashMap<>();
        fields.put("v1", Double.parseDouble(objectList.get(i++).toString()));
        fields.put("v2", Double.parseDouble(objectList.get(i++).toString()));
        fields.put("v3", Double.parseDouble(objectList.get(i).toString()));

        return new InfluxDBPoint(tbl, timestamp, tags, fields);
    }
}
